package com.jy.oms.framework.asyn.task.queue.database;

import com.jy.oms.framework.asyn.task.constants.CacheConstants;
import com.jy.oms.framework.asyn.task.constants.TaskQueueTypeEnum;
import com.jy.oms.framework.asyn.task.core.ExecutorConfig;
import com.jy.oms.framework.asyn.task.core.ExecutorContext;
import com.jy.oms.framework.asyn.task.core.RedissonClientProxy;
import com.jy.oms.framework.asyn.task.handler.ITaskHandlerManager;
import com.jy.oms.framework.asyn.task.handler.ITaskHandlerProcessor;
import com.jy.oms.framework.asyn.task.model.*;
import com.jy.oms.framework.asyn.task.queue.ITaskQueueManager;
import com.jy.oms.framework.asyn.task.queue.database.entity.SysTaskQueue;
import com.jy.oms.framework.asyn.task.queue.database.service.ISysTaskQueueService;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.redisson.api.RBucket;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * db队列任务统计代理
 * 1. 添加任务时，标记对应任务组有数据
 * 2. 查询任务时，判断相对应任务组是否有任务数据标记，如果无，则直接返回
 *
 * @author liuquan
 */
@Service
public class DbTaskQueueStatisticManager implements ITaskQueueManager, ITaskHandlerProcessor {

    private static final long EXPIRE_TIME = 10;
    @Autowired
    private DbTaskQueueManager dbTaskQueueManager;
    @Autowired
    private ISysTaskQueueService sysTaskQueueService;
    @Autowired
    protected ITaskHandlerManager taskHandlerManager;

    private RBucket<Integer> getBucket(String taskGroup) {
        String cacheKey = CacheConstants.genCacheKey(CacheConstants.TASK_STATISTIC, taskGroup);
        RedissonClient redissonClient = RedissonClientProxy.getRedissonClient();
        RBucket<Integer> bucket = redissonClient.getBucket(cacheKey);
        return bucket;
    }

    private void summaryTask(String taskGroup) {
        if (StringUtils.isBlank(taskGroup)) {
            return;
        }

        RBucket<Integer> bucket = this.getBucket(taskGroup);
        bucket.set(1, EXPIRE_TIME, TimeUnit.MINUTES);
    }

    @Override
    public void add(TaskWrapper taskWrapper) {
        dbTaskQueueManager.add(taskWrapper);

        if (!ExecutorContext.isDevOrTest()) {
            this.summaryTask(taskWrapper.getTaskGroup());
        }
    }

    @Override
    public void addBatch(List<TaskWrapper> taskWrapperList) {
        dbTaskQueueManager.addBatch(taskWrapperList);

        if (!ExecutorContext.isDevOrTest()) {
            taskWrapperList.stream().map(TaskWrapper::getTaskGroup).forEach(this::summaryTask);
        }
    }

    @Override
    public List<TaskWrapper> poll(TaskQueryDto taskQueryDto) {
        if (ExecutorContext.isDevOrTest()) {
            List<TaskWrapper> taskWrapperList = dbTaskQueueManager.poll(taskQueryDto);
            return taskWrapperList;
        } else {
            RBucket<Integer> bucket = this.getBucket(taskQueryDto.getTaskGroup());
            Integer checkSummaryFlag = bucket.get();
            if (checkSummaryFlag != null && checkSummaryFlag == 0) {
                return Collections.emptyList();
            }

            List<TaskWrapper> taskWrapperList = dbTaskQueueManager.poll(taskQueryDto);

            ExecutorConfig executorConfig = ExecutorContext.getExecutorConfig();
            if (CollectionUtils.isEmpty(taskWrapperList)
                    // 集群环境比较复杂，暂不考虑
                    && !executorConfig.isEnableCluster()) {
                if (checkSummaryFlag == null) {
                    //场景：查询DB后在极短时间内有添加任务
                    bucket.compareAndSet(null, 0);
                } else {
                    bucket.compareAndSet(0, 0);
                }
                bucket.expire(EXPIRE_TIME, TimeUnit.MINUTES);
            }

            return taskWrapperList;
        }
    }

    @Override
    public void finish(TaskResultDto taskResultDto) {
        dbTaskQueueManager.finish(taskResultDto);
    }

    @Override
    public void finish(List<TaskResultDto> taskResultList) {
        dbTaskQueueManager.finish(taskResultList);
    }

    @Override
    public void remove(TaskWrapper taskWrapper) {
        dbTaskQueueManager.remove(taskWrapper);
    }

    /**
     * 记录db队列表名对应的代理统计key
     */
    @Override
    public void processAfterTaskHandlerInitialization() {
        Map<TaskGroup, List<TaskHandlerWrapper>> taskGroupListMap = taskHandlerManager.loadAllExecutableTaskHandler();
        List<Pair<String, String>> taskQueueNameMappings = taskGroupListMap.entrySet().stream()
                .filter(entry -> entry.getKey().getTaskQueueType() == TaskQueueTypeEnum.DB)
                .filter(entry -> CollectionUtils.isNotEmpty(entry.getValue()))
                .map(entry -> {
                    String taskGroupDbQueueName = entry.getValue().get(0).getExecuteStrategy().getTaskGroupDbQueueName();
                    if (StringUtils.isBlank(taskGroupDbQueueName)) {
                        return null;
                    }
                    String taskGroup = entry.getKey().getGroup();
                    String statisticKey = CacheConstants.genCacheKey(CacheConstants.TASK_STATISTIC, taskGroup);
                    return Pair.of(taskGroupDbQueueName, statisticKey);
                })
                .filter(Objects::nonNull)
                .collect(Collectors.toList());
        if (CollectionUtils.isEmpty(taskQueueNameMappings)) {
            return;
        }

        RedissonClient redissonClient = RedissonClientProxy.getRedissonClient();
        for (Pair<String, String> mapping : taskQueueNameMappings) {
            String cacheKey = CacheConstants.genFrameworkKey(CacheConstants.QUEUE_STATISTIC_CACHE_MAPPING, mapping.getLeft());
            RBucket<String> bucket = redissonClient.getBucket(cacheKey);
            bucket.set(mapping.getRight());
        }
    }

    /**
     * 由前端页面触发重跑任务
     */
    public void resetTask(SysTaskQueue sysTask, String taskQueueName) {
        if (sysTask == null || sysTask.getId() == null) {
            return;
        }

        sysTaskQueueService.updateById(sysTask, StringUtils.trimToEmpty(taskQueueName));

        if (StringUtils.isBlank(taskQueueName)) {
            this.summaryTask(sysTask.getTaskGroup());
        } else {
            RedissonClient redissonClient = RedissonClientProxy.getRedissonClient();
            String cacheKey = CacheConstants.genFrameworkKey(CacheConstants.QUEUE_STATISTIC_CACHE_MAPPING, taskQueueName);
            RBucket<String> bucket = redissonClient.getBucket(cacheKey);
            String statisticCacheKey = bucket.get();
            if (StringUtils.isBlank(statisticCacheKey)) {
                return;
            }

            RBucket<Integer> statisticBucket = redissonClient.getBucket(statisticCacheKey);
            if (statisticBucket.isExists()) {
                statisticBucket.set(1, EXPIRE_TIME, TimeUnit.MINUTES);
            }
        }
    }

}
